package com.minivision.fdiot.mqtt.core;

import java.io.IOException;
import java.nio.ByteBuffer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PathVariable;

import com.fasterxml.jackson.core.JsonParser;
import com.minivision.fdiot.mqtt.MessageContext;
import com.minivision.fdiot.mqtt.RequestFuture;
import com.minivision.fdiot.mqtt.core.MqttMessageHandler.MqttMessageParam;
import com.minivision.fdiot.mqtt.core.MqttMessageHandler.TopicHandler;
import com.minivision.fdiot.mqtt.core.MqttMessageHandler.MqttMessageParam.ParamType;
import com.minivision.fdiot.protopool.Packet;
import com.minivision.fdiot.protopool.PacketUtils;
import com.minivision.fdiot.protopool.Packet.Head;
import com.minivision.fdiot.protopool.Packet.Head.Type;

import io.netty.buffer.ByteBuf;
import lombok.extern.slf4j.Slf4j;

/**
 * topic处理器
 * @author hughzhao
 * @2018年3月1日
 */
@Slf4j
@MqttMessageHandler
public class TopicMessageHandler {

  @Autowired
  private PacketUtils packetUtils;
  
  @Autowired
  private MessageContext messageContext;
  
  @Autowired
  private CmdMessageDeliver codeMessageDeliver;
  
  /**
   * json消息处理
   * @param clientId
   * @param username
   * @param payload
   * @param model
   */
  @TopicHandler("/s/{model}")
  public void msgProcess(@MqttMessageParam(ParamType.clientId) String clientId, @MqttMessageParam(ParamType.username) String username, ByteBuf payload, @PathVariable("model") String model) {
    try {
      byte[] payloadBytes = new byte[payload.readableBytes()];
      payload.getBytes(0, payloadBytes);
      if (log.isDebugEnabled()) {
        log.debug("receive raw message:{}", new String(payloadBytes, "utf-8"));
      }
      JsonParser parser = packetUtils.createParser(payloadBytes);
      Head head = packetUtils.parseHead(parser);
      log.info("receive a packet, head:{}", head);
      int code = head.getCode();
      int type = head.getType();
      
      if(Type.isResponse(type)){
        handleGenericResponse(head, parser);
        return;
      }
      codeMessageDeliver.deliver(code, type, model, clientId, username, head, parser);
    } catch (IOException e) {
      log.error("deserialize payload error", e);
    } finally {
      payload.release();
    }
  }
  
  /**
   * 二进制消息处理
   * @param clientId
   * @param username
   * @param payload
   * @param model
   */
  @TopicHandler("/sb/{model}")
  public void binMsgProcess(@MqttMessageParam(ParamType.clientId) String clientId, @MqttMessageParam(ParamType.username) String username, ByteBuf payload, @PathVariable("model") String model) {
    try {
      byte[] payloadBytes = new byte[payload.readableBytes()];
      payload.getBytes(0, payloadBytes);
      ByteBuffer buffer = ByteBuffer.wrap(payloadBytes);
      Head head = packetUtils.parseHead(buffer);
      log.info("receive a packet, head:{}", head);
      int code = head.getCode();
      int type = head.getType();
      buffer.position(PacketUtils.BIN_HEAD_LEN);
      
      byte[] bodyBytes = new byte[buffer.remaining()];
      buffer.get(bodyBytes);
      if(Type.isResponse(type)){
        handleGenericResponse(head, bodyBytes);
        return;
      }
      codeMessageDeliver.deliver(code, type, model, clientId, username, head, bodyBytes);
    } catch (IOException e) {
      log.error("deserialize payload error", e);
    } finally {
      payload.release();
    }
  }
  
  @SuppressWarnings({"rawtypes", "unchecked"})
  private void handleGenericResponse(Head h, Object bodyWrapper) throws IOException {
    RequestFuture<?> req = messageContext.remove(h.getId());
    if (req == null) {
        log.error("No request found for a response, maybe timeout, just discard it. Head : {}", h);
        return;
    }
    try {
        Object body = packetUtils.parseBody(bodyWrapper, req.getResponseBodyType());
        Packet response = new Packet<>(h, body);
        req.setResponse(response);
    } catch (Exception e) {
        log.error("parseBody error, head:" + h, e);
        req.fail(e);
    }
  }
  
  
  /*@SuppressWarnings({"rawtypes", "unchecked"})
  private void handleGenericResponse(Head h, byte[] bodyBytes) throws IOException {
    RequestFuture<?> req = messageContext.remove(h.getId());
    if (req == null) {
        log.error("No request found for a response, maybe timeout, just discard it. Head : {}", h);
        return;
    }
    try {
        Object body = packetUtils.parseBody(bodyBytes, req.getResponseBodyType());
        Packet response = new Packet<>(h, body);
        req.setResponse(response);
    } catch (Exception e) {
        log.error("parseBody error, head: {}", h, e);
        req.fail(e);
    }
  }*/
  
}
